LangGraph速记-开发者社区

一 Langgraph orientation（出发点）

1 Latency in the seconds vs ms（通过并行/流式输出减少延迟以及感官延迟）

Parallelization – to save actual latency
Streaming – to save perceived latency

2 Long-Running Agents can fail, which is expensive and time consuming（通过记忆解决工程上的每一次从头开始）

Checkpointing – to reduce the cost of each retry

3 The non-deterministic nature of AI requires checkpoints, approvals, and testing（通过interupt 解决人机交互）

Human-in-the-loop - to collaborate with the user
Tracing, Observation and Evaluation (LangSmith)

二 Langgraph Components(基础组建)

1. State&Nodes：

State seems like the thread data that I’ve been familiar with，can be persisted across time and in particular across failures of nodes，是一个流动的状态数据。

We get 3 ways to define the state:

TypeDict
dataclass
Pydantic Model

Nodes：The specific function that we define in graph, which is used for solving certain problems.
下面是基础用法：

fromdataclassesimportdataclass,fieldfrompydanticimportBaseModel,FieldfromtypingimportAnnotated,List,Literal,TypedDictfromlanggraph.graphimportEND,START,StataGraphfromlanggraph.typesimportCommand,interruptfromtypingimportTypedDict,ListclassState(TypedDict):messages:List[str]todo:List[str]approved:bool@dataclassclassState:messages:List[str]=field(default_factory=list)todo:List[str]=field(default_factory=list)approved:bool=FalseclassState(BaseModel):messages:List[str]=Field(default_factory=list)todo:List[str]=Field(default_factory=list)approved:bool=Falsedefplan_node(state:State):# 假设根据用户输入做一个计划return{"todo":["search docs","summarize","draft blog"],"messages":state["messages"]+["[plan] generated todo list"]}defwork_node(state:State):done=f"[work] did:{state['todo'][0]}"return{"todo":state["todo"][1:],"messages":state["messages"]+[done]}

2. Edges: Control Flow

核心概念：Super Steps

Super Step 是 LangGraph 保证「并行安全、状态一致、流程可控」的核心执行语义。

Super Step = 一轮「节点并行执行 → 状态统一合并 → 再推进流程」的原子步骤

它不是一个函数，也不是一个类，而是LangGraph 在内部推进 Graph 时的“节拍器, 同一个 Super Step 内，所有节点看到的是“同一份旧 State”,

顺序边

fromlanggraph.graphimportStateGraph,START,END builder=StateGraph(State)builder.add_node("plan",plan_node)builder.add_node("work",work_node)builder.add_edge(START,"plan")builder.add_edge("plan","work")builder.add_edge("work",END)

Parallel

可以控制并行执行加速

defresearch_a(state:State):return{"messages":state["messages"]+["[A] research result"]}defresearch_b(state:State):return{"messages":state["messages"]+["[B] research result"]}defmerge(state:State):return{"messages":state["messages"]+["[merge] merged A+B"]}builder=StateGraph(State)builder.add_node("plan",plan_node)builder.add_node("research_a",research_a)builder.add_node("research_b",research_b)builder.add_node("merge",merge)builder.add_edge(START,"plan")# 并行分叉builder.add_edge("plan","research_a")builder.add_edge("plan","research_b")# 汇合（两个都跑完后进入 merge）builder.add_edge("research_a","merge")builder.add_edge("research_b","merge")builder.add_edge("merge",END)

进一步的，可以使用map-reduce

langgraph用Annotated定义 reducer

fromtypingimportAnnotated,List,TypedDictfromlanggraph.graphimportStateGraph,START,ENDfromoperatorimportaddclassState(TypedDict):# 多个并行节点都可以往 results 里写# 最终用 add（list concat）做 reduceresults:Annotated[List[str],add]builder=StateGraph(State)builder.add_node("map_a",map_a)builder.add_node("map_b",map_b)builder.add_node("map_c",map_c)# fan-outbuilder.add_edge(START,"map_a")builder.add_edge(START,"map_b")builder.add_edge(START,"map_c")# fan-in（隐式）builder.add_edge("map_a",END)builder.add_edge("map_b",END)builder.add_edge("map_c",END)

Let get into what’s happened under the hook：
1）开始state

{"results":[]}

2）Super Step 1（并行 Map）

map_a →{"results": ["result from A"]}
map_b →{"results": ["result from B"]}
map_c →{"results": ["result from C"]}

3）最后：

{"results":["result from A","result from B","result from C"]}

问题：什么时候该用 Map-Reduce，而不是 merge node？

✅ 用 State reducer（Map-Reduce）

并行节点彼此独立
只是“收集结果”
合并逻辑简单（list / sum / dict merge）

✅ 用显式 merge node

合并逻辑复杂
需要排序、过滤、再推理
合并本身是一个“智能步骤”

condition controll

defroute_after_work(state:State):# todo 还有就继续 work，否则结束return"work"ifstate["todo"]elseEND builder=StateGraph(State)builder.add_node("plan",plan_node)builder.add_node("work",work_node)builder.add_edge(START,"plan")builder.add_edge("plan","work")builder.add_conditional_edges("work",route_after_work,["work",END])

3 Memory(checkpoint)

执行中间结果的缓存。一个非常关键的观点: A thread is a collection of checkpoints

一个 thread 本质上是一组 checkpoints。

Checkpoint 的意义不只是失败恢复，还包括：

调试（复盘每一步 State）
评估（分析哪个阶段出问题）
治理（知道 Agent 实际做过什么）

这一步，决定了 Agent 是否能进入生产环境。

fromlanggraph.checkpoint.memoryimportInMemorySaver memory=InMemorySaver()graph=builder.compile(checkpointer=memory)config={"configurable":{"thread_id":"thread-1"}}initial_state={"messages":["hi"],"todo":[],"approved":False}final_state=graph.invoke(initial_state,config=config)print(final_state["messages"])

4. Human in the Loop

LangGraph 并不假设 AI 可以全自动完成任务。

相反，它默认：

某些节点必须人工确认
某些决策应该暂停等待人类
某些路径需要人工选择

这让 Agent 更像一个：

人机协作流程系统，而不是 autonomous bot。

fromlanggraph.typesimportinterruptdefapproval_node(state:State):# 中断，把当前关键信息抛给人decision=interrupt({"question":"Approve this plan?","todo":state["todo"],"messages_tail":state["messages"][-3:]})# decision 是人工返回的内容（比如 {"approved": True} 或 {"approved": False, "fix": "..."}）return{"approved":bool(decision.get("approved",False)),"messages":state["messages"]+[f"[human] decision={decision}"]}builder=StateGraph(State)builder.add_node("plan",plan_node)builder.add_node("approval",approval_node)builder.add_node("work",work_node)builder.add_edge(START,"plan")builder.add_edge("plan","approval")defroute_after_approval(state:State):return"work"ifstate["approved"]elseEND builder.add_conditional_edges("approval",route_after_approval,["work",END])builder.add_conditional_edges("work",lambdas:"work"ifs["todo"]elseEND,["work",END])graph=builder.compile(checkpointer=InMemorySaver())