LangChain / LangGraph 学习笔记：从链式调用到状态机 Agent 运行时

写作主线：LangChain 的演进本质是编排权的三次升级——从「封装一次模型调用」到「声明式管道（LCEL）」，再到「用状态图表达循环/分支/持久化/人机协同的 Agent 运行时（LangGraph）」，最后在 1.x 用统一的 create_agent 收口。 每个知识点讲 是什么 → 为什么这么设计 → 怎么实现（最小可运行代码）。代码栈 Python，已用 langchain 1.4 / langgraph 1.2 / deepseek-v4-flash 实测，配套 langchain_langgraph_demo.ipynb。

学习笔记的配套代码： https://github.com/LT-IENG/AI-Agent-Study-Notes

---

目录

• 0. 导读与版本说明
• 1. 背景：为什么需要 LLM 应用框架
• [2. LangChain 生态全景与分层](#2-langchain 生态全景与分层)
• 3. 核心抽象一：Model I/O（模型 / 提示词 / 输出解析）
• 4. 核心抽象二：Runnable 协议与 LCEL 声明式编排
• 5. 核心抽象三：工具调用（Function/Tool Calling）
• 6. 旧 API 的演进与弃用（重要避坑）
• 7. 为什么会有 LangGraph：线性链的五个做不到
• 8. LangGraph 核心机制（重点）
• 9. 持久化、记忆与人在回路
• 10. 多 Agent 编排模式
• 11. 1.x 的统一高层 API：create_agent
• 12. LangSmith：让链路可观测、可评测
• 13. LangChain vs LangGraph：选型与最佳实践
• 14. 参考资料

---

0. 导读与版本说明

先建立一张演进地图，后面所有内容都是它的展开：

flowchart LR
    A["① 裸 SDK<br/>每次手写请求/解析/重试"] --> B["② LangChain 组件<br/>Prompt/Model/Parser 抽象"]
    B --> C["③ LCEL 管道<br/>用 | 声明式串联，自动流式/并行/重试"]
    C --> D["④ 旧 Agent(AgentExecutor)<br/>黑盒循环，难控制"]
    D --> E["⑤ LangGraph<br/>状态图：循环/分支/持久化/人机协同"]
    E --> F["⑥ 1.x create_agent<br/>统一的现代 Agent 运行时"]

版本提示（非常重要，网上大量教程已过时）：本笔记基于 langchain 1.x / langgraph 1.x。

• LLMChain、AgentExecutor、ConversationBufferMemory 等旧 API 已弃用，进入 langchain-classic 兼容层，新项目不要用（第 6 章详解）。
• langgraph.prebuilt.create_react_agent 在 LangGraph 1.0 起迁移为 langchain.agents.create_agent（旧名仍可用但有弃用告警，将在 2.0 移除）。
• langchain-community 正在被拆分为独立集成包并逐步 sunset，用到时知道去哪找即可。

---

1. 背景：为什么需要 LLM 应用框架

1.1 裸用模型 SDK 会反复造的轮子

直接用 HTTP/SDK 调一次大模型很简单，但要把它做成应用，你会反复手写这些东西：

• 把「系统指令 + 变量 + 历史 + 检索结果」拼成 Prompt（到处是字符串拼接和转义）；
• 把模型返回的文本解析成结构化对象（正则、JSON 容错）；
• 超时重试、限流退避、缓存、批量、流式输出；
• 多步流程：上一步输出喂给下一步、有的步骤要并行、有的要按结果分支；
• 接入外部工具（Function Calling 的参数校验、结果回填）；
• 多轮对话的记忆、断线恢复、人工审批；
• 全链路日志、追踪、评测。

LangChain/LangGraph 的价值就是把这些共性能力标准化：提供统一抽象（组件可替换）、声明式编排（少写胶水代码）、以及一个能支撑生产级 Agent 的运行时。

1.2 一个直观对比：同一个「提示词→模型→解析」任务

裸 SDK 写法（每次都要手动拼 messages、手动处理返回）：

# 裸 OpenAI SDK
from openai import OpenAI
client = OpenAI()
resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user",
               "content": f"把下面这句话翻译成英文：{text}"}],
)
result = resp.choices[0].message.content  # 还要自己做解析/重试/流式...

LangChain 写法（组件可替换、一行管道、自带流式/重试）：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = (
    ChatPromptTemplate.from_template("把下面这句话翻译成英文：{text}")
    | llm
    | StrOutputParser()      # 直接拿字符串，不用手动抠 choices
)
print(chain.invoke({"text": "你好世界"}))     # 同步
for tok in chain.stream({"text": "你好世界"}):  # 同一套接口直接流式
    print(tok, end="")

关键差异不是「少打几个字」，而是：模型可随意替换（OpenAI/DeepSeek/本地模型同一接口）、每个环节都是标准组件、invoke/batch/stream/ainvoke 四种调用方式自动获得。

---

2. LangChain 生态全景与分层

LangChain 不是一个包，而是一套分层生态，理解分层就不会在 import 时迷路：

flowchart TD
    APP[你的应用] --> LS[LangSmith 可观测/评测/Prompt管理（平台，可选）]
    APP --> LG[LangGraph：Agent 运行时（状态图/持久化/人机协同/多Agent）]
    LG --> LC[langchain：高层 API、链、Agent 封装（1.x 含 create_agent）]
    LC --> CORE[langchain-core：核心抽象 Runnable/Message/Prompt/Tool]
    LC --> PARTNER[langchain-openai / -anthropic ... 各模型集成包]
    LC --> COMM[langchain-community：社区集成（逐步拆分/sunset）]

包	职责	你主要用它的什么
langchain-core	最底层抽象，不依赖具体模型	Runnable、消息类型、PromptTemplate、tool、OutputParser
langchain-openai / -anthropic 等	具体厂商模型的对接（OpenAI 兼容接口都用 langchain-openai）	ChatOpenAI、OpenAIEmbeddings
langchain	高层封装：检索链、Agent、文档加载等	1.x 的 create_agent、检索相关组件
langchain-community	大量第三方集成（向量库、加载器、SQLDatabase 等）	找冷门集成，注意其逐步 sunset
langgraph	Agent / 工作流运行时	StateGraph、checkpointer、interrupt
langsmith（平台）	追踪、评测、数据集、Prompt 管理	debug 与线上监控（第 12 章）

设计哲学：core 定义标准接口（稳定、少变），具体实现放到 partner/community 包（可频繁迭代、可替换），上层 langchain/langgraph 负责编排。这样换模型、换向量库时业务代码几乎不动。

---

3. 核心抽象一：Model I/O（模型 / 提示词 / 输出解析）

Model I/O 是一切的基础，对应经典三段式：Prompt（格式化输入）→ ChatModel（调用模型）→ OutputParser（解析输出）。

flowchart LR
    V[变量 dict] --> P[ChatPromptTemplate 格式化]
    P --> M[ChatModel 调用]
    M --> O[OutputParser 解析为对象]

3.1 消息类型：和 Chat API 一一对应

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
# SystemMessage：设定角色/规则；HumanMessage：用户；AIMessage：模型回复（可含 tool_calls）；
# ToolMessage：工具执行结果回填。多轮对话本质就是一个 message 列表。
messages = [
    SystemMessage("你是简洁的翻译官"),
    HumanMessage("翻译：早上好"),
]

3.2 ChatPromptTemplate：变量化、可组合的提示词

from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}，回答不超过{n}字"),
    ("human", "{question}"),
])
# invoke 时填变量，输出标准 Message 列表
print(prompt.invoke({"role": "数据库专家", "n": 50, "question": "什么是索引？"}).to_messages())

3.3 OutputParser：把自由文本变成程序可用的对象

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

StrOutputParser()                 # 最常用：直接取 .content 字符串
JsonOutputParser()                # 让模型输出 JSON 并解析成 dict（配合 pydantic 更稳）

# 更推荐的现代做法：用 Pydantic 做强类型约束（结构化输出的底层就是一次工具调用）
from pydantic import BaseModel
class City(BaseModel):
    city: str
    country: str
# 通用写法：bind_tools 直接传 Pydantic 类，默认 tool_choice=auto，兼容推理型模型
ai = llm.bind_tools([City]).invoke("法国首都是哪里？通过调用工具返回城市和国家")
city = City(**ai.tool_calls[0]["args"])   # -> City(city='巴黎', country='法国')
# OpenAI 官方模型也可用更短的 llm.with_structured_output(City)；
# 但它默认强制 tool_choice / json_schema response_format，DeepSeek 等推理/兼容模型可能报错，
# 此时用上面 bind_tools 的等价写法最稳。
print(city)

为什么推荐 with_structured_output 而不是让模型吐 JSON 再正则解析？ 因为它借助模型原生的 Function Calling 机制返回带 schema 约束、类型可校验的结果，避免了「JSON 多一个逗号、被 markdown 包裹、字段缺失」等无尽的解析容错。

---

4. 核心抽象二：Runnable 协议与 LCEL 声明式编排

4.1 是什么：一切皆 Runnable，用 | 拼管道

LCEL（LangChain Expression Language） 是 LangChain 的声明式编排语法。核心是 Runnable 协议：Prompt、ChatModel、Parser、Retriever、自定义函数……全部实现同一套接口，因此可以用管道符 | 像 Unix 管道一样串起来，前一个的输出自动作为后一个的输入。

每个 Runnable 都有四种调用方式（这就是「协议」的价值——学一次到处用）：

方法	作用
invoke(input)	同步调用一次
ainvoke(input)	异步调用一次
batch([...])	批量（内部自动并发）
stream(input)	流式逐 token / 逐块产出

4.2 为什么这么设计：声明式 > 命令式胶水代码

如果用命令式手写「检索→拼 prompt→调模型→解析」，每步都要手动传参、处理并行和流式；LCEL 只声明「数据怎么流」，框架自动获得：流式（逐块透传）、并行（RunnableParallel）、重试/超时（RunnableConfig）、批处理、回调与 LangSmith 追踪。

4.3 关键编排原语（最小代码）

from langchain_core.runnables import (
    RunnablePassthrough, RunnableLambda, RunnableParallel, RunnableBranch
)

# ① RunnablePassthrough：原样透传；.assign 可在不丢原字段的情况下新增字段
chain = RunnablePassthrough.assign(
    greeting=lambda x: f"你好，{x['name']}"
)
chain.invoke({"name": "小明"})   # {'name': '小明', 'greeting': '你好，小明'}

# ② RunnableLambda：把任意普通函数包成 Runnable，接入管道
upper = RunnableLambda(lambda x: x.upper())
upper.invoke("hello")            # HELLO

# ③ RunnableParallel：多路并行执行，最后合并成 dict（自动并发）
parallel = RunnableParallel(
    summary=ChatPromptTemplate.from_template("一句话总结：{text}") | llm | StrOutputParser(),
    keywords=ChatPromptTemplate.from_template("提取3个关键词，逗号分隔：{text}") | llm | StrOutputParser(),
    original=RunnablePassthrough(),   # 原文也保留
)
parallel.invoke({"text": "..."})      # {'summary':..,'keywords':..,'original':..}

# ④ 经典 RAG 链：检索结果用 retriever 自动填充 context，问题原样透传
rag_chain = (
    RunnablePassthrough.assign(
        context=lambda x: "\n".join(d.page_content for d in retriever.invoke(x["question"]))
    )
    | ChatPromptTemplate.from_template("根据资料回答：\n{context}\n问题：{question}")
    | llm
    | StrOutputParser()
)

4.4 RunnableBranch：声明式条件分支

branch = RunnableBranch(
    (lambda x: "投诉" in x["text"], ChatPromptTemplate.from_template("安抚并转人工：{text}") | llm),
    (lambda x: "咨询" in x["text"], ChatPromptTemplate.from_template("专业解答：{text}") | llm),
    ChatPromptTemplate.from_template("常规回复：{text}") | llm,   # 默认分支
)

4.5 LCEL 的能力边界（关键，引出 LangGraph）

LCEL 是 DAG（有向无环图）：数据只能向前流，不能回头循环、不能在中途暂停等人、不能跨步骤维护可变状态、不能动态决定走几步。它非常适合「一次性流过」的链（RAG、翻译、摘要），但 Agent 的本质是「思考-行动-观察的循环」，这超出了 DAG 的表达能力——这正是 LangGraph 要解决的问题（第 7、8 章）。

---

5. 核心抽象三：工具调用（Function/Tool Calling）

5.1 是什么与为什么

Tool Calling（函数调用）：模型不直接给最终答案，而是输出一个结构化的「我要调用哪个工具、参数是什么」，由你的代码真正执行工具、把结果回填，模型再据此继续。它是 Agent 能「动手做事」的基石。之所以用结构化 schema 而非让模型输出自由文本，是为了让程序能可靠解析、校验参数。

5.2 三种定义工具的方式

from langchain_core.tools import tool

# 方式 1：@tool 装饰器（最常用）。docstring 会成为给模型看的“使用说明”，类型注解决定参数 schema
@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积。当用户需要乘法时使用。"""
    return a * b

# 方式 2：StructuredTool / 从 Pydantic 定义（需要复杂参数校验时）
print(multiply.name, multiply.description)           # multiply / 计算两个整数...
print(multiply.args)                                # 自动生成的 JSON Schema
multiply.invoke({"a": 3, "b": 4})                   # 程序侧直接执行 -> 12

5.3 模型侧的工具调用长什么样

llm_with_tools = llm.bind_tools([multiply])          # 把工具 schema 绑定给模型
ai_msg = llm_with_tools.invoke("3 乘 4 等于多少")
print(ai_msg.tool_calls)
# [{'name': 'multiply', 'args': {'a': 3, 'b': 4}, 'id': 'call_xxx', 'type': 'tool_call'}]

一次完整的工具调用闭环（手写一遍，胜过看十张图）：

from langchain_core.messages import HumanMessage, ToolMessage

messages = [HumanMessage("3 乘 4 等于多少")]
ai = llm.bind_tools([multiply]).invoke(messages)
messages.append(ai)
for tc in ai.tool_calls:                                  # 模型决定要调的工具
    result = {"multiply": multiply}[tc["name"]].invoke(tc["args"])  # 你的代码真正执行
    messages.append(ToolMessage(str(result), tool_call_id=tc["id"])) # 结果回填，id 必须对应
final = llm.invoke(messages)                             # 模型拿到工具结果给最终答案
print(final.content)

两个易错点：① ToolMessage.tool_call_id 必须与请求的 id 对应，多工具并行时尤其重要；② 工具的 docstring 和参数描述是给模型看的，写不清模型就会调错——这在 Agent Harness 笔记里被称为 ACI（Agent-Computer Interface）设计。

---

6. 旧 API 的演进与弃用（重要避坑）

网上海量 LangChain 教程停留在 2023–2024 的旧写法，混用会让项目非常痛苦。这张对照表请直接收藏：

能力	❌ 旧写法（已弃用，在 langchain-classic 兼容层）	✅ 现代写法（1.x）
简单链	LLMChain(llm=, prompt=)	`prompt
Agent 循环	AgentExecutor、initialize_agent	LangGraph / langchain.agents.create_agent
对话记忆	ConversationBufferMemory 等 Memory 类	Checkpointer（按 thread 持久化消息）
检索问答	RetrievalQA.from_chain_type	LCEL：`retriever
ReAct 预置	create_react_agent（langgraph.prebuilt，1.0 起弃用告警）	from langchain.agents import create_agent
回调驱动	大量 CallbackHandler	LangSmith trace + LCEL 内置事件

为什么要弃用旧 API（设计动机）：

1. LLMChain 太死板：它把「prompt+llm+parser」封成黑盒，无法灵活组合并行、分支、流式，LCEL 的 Runnable 组合性更强。
2. AgentExecutor 难控制：黑盒循环，难以插入循环上限、持久化、人工审批、子图、多 Agent；LangGraph 把控制权显式交给开发者。
3. Memory 类和链耦合：旧 Memory 绑死在特定链上、无法跨会话/断线恢复；Checkpointer 用「状态快照」统一解决短期记忆、断点续跑和时间旅行。

结论：看到教程里出现 LLMChain / AgentExecutor / initialize_agent / ConversationBufferMemory，知道它在讲什么即可，新代码一律用右侧现代写法。

---

7. 为什么会有 LangGraph：线性链的五个做不到

LCEL 的 DAG 足够搭「一次性流过」的应用，但真正的 Agent 需要：

1. 循环（Cycles）：ReAct 是「思考→行动→观察」反复进行，走几步由运行时决定，DAG 无法表达环。
2. 条件分支与动态路由（Branching）：根据中间结果决定下一步去哪，甚至回到上一步重试。
3. 可变共享状态（State）：多步之间要读写同一份上下文（消息列表、中间产物、重试计数）。
4. 持久化与断点续跑（Persistence）：进程崩了能从上次的状态恢复；多轮对话要记住历史。
5. 人在回路（Human-in-the-loop）：在危险操作前暂停、等人审批/补充信息后再继续。
6. 多 Agent 协作：多个角色 Agent 之间分工、交接、汇总。

LangGraph 的答案是：把 Agent 应用建模成一张「状态图（StateGraph）」——节点（Node）是计算步骤，边（Edge）是流转规则，一份全局 State 在节点间流动并被增量更新。它借鉴了 Google Pregel / Apache Beam 的 super-step（超步） 思想：每个超步里节点并行计算、通过 reducer 合并状态、投票决定是否进入下一步。

flowchart LR
    subgraph 线性链 LCEL
        l1[步骤1] --> l2[步骤2] --> l3[步骤3]
    end
    subgraph 状态图 LangGraph
        s1[节点A] --> c{条件边}
        c -->|情况1| s2[节点B]
        c -->|情况2| s3[节点C]
        s2 --> s1
        s3 --> s4[节点D]
    end

---

8. LangGraph 核心机制（重点）

8.1 三大基本要素：State / Node / Edge

• State（状态）：一个 TypedDict，是贯穿全图的共享数据。
• Node（节点）：普通函数 def node(state) -> dict，读取 state、只返回要更新的字段（增量）。
• Edge（边）：普通边固定跳转；条件边（conditional_edges） 根据 state 动态决定下一节点。

8.2 Reducer：State 字段怎么合并（最容易踩坑的点）

节点返回的是「增量」，多个节点（或并行节点）更新同一字段时如何合并，由该字段的 Reducer 决定：

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

class State(TypedDict):
    question: str                       # 不写 reducer：默认“覆盖”
    steps: Annotated[list, operator.add]  # operator.add：列表“追加”而非覆盖
    messages: Annotated[list, add_messages]  # 消息专用：按 message id 合并/更新

Reducer	行为	典型用途
不指定	后写覆盖先写	问题、最终答案等单值
operator.add	列表拼接追加	累积中间步骤、工具结果
add_messages	按 ID 智能合并消息	对话消息列表（更新同 id、追加新消息）

为什么需要 reducer：图可能并行运行多个节点，若都返回 steps=[...]，默认覆盖会互相丢数据；声明 operator.add 才能安全地把各自结果累加。这是从「命令式赋值」转向「状态归并」的关键思维。

8.3 最小可运行状态图

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    n: int
    log: Annotated[list, operator.add]

def incr(s):  return {"n": s["n"] + 1, "log": ["+1"]}
def double(s): return {"n": s["n"] * 2, "log": ["*2"]}
def check(s): return "double" if s["n"] < 5 else "end"   # 条件路由

g = StateGraph(S)
g.add_node("incr", incr)
g.add_node("double", double)
g.add_edge(START, "incr")
g.add_conditional_edges("incr", check, {"double": "double", "end": END})
g.add_edge("double", "incr")      # 注意：double 又指回 incr，形成“循环”
app = g.compile()

print(app.invoke({"n": 0, "log": []}))   # 图会循环直到 n>=5

START/END 是虚拟节点，表示入口和结束。compile() 后得到一个标准 Runnable，同样支持 invoke/stream/ainvoke，并可用 app.get_graph().draw_mermaid_png() 出图。

8.4 循环护栏：recursion_limit

图能循环，也就可能死循环。LangGraph 默认限制超步数量（约 25），超限抛 GraphRecursionError，也可在 config 里设置：

app.invoke(state, config={"recursion_limit": 20})

8.5 从零手写一个 ReAct Agent（拆开 create_agent 黑盒）

理解 Agent 运行时最好的方式，是不用任何预置、用 StateGraph 亲手实现 ReAct 循环。核心就两个节点：agent（模型决定下一步）和 tools（执行工具），一条条件边判断「模型还要调工具 → 回 tools，否则 → END」。

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

tools = [multiply]                                   # 第 5 章定义的工具
tools_by_name = {t.name: t for t in tools}
llm_with_tools = llm.bind_tools(tools)

def agent_node(state):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def tools_node(state):
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:                       # 支持一次并行调多个工具
        out = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(str(out), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue(state):
    return "tools" if state["messages"][-1].tool_calls else END

g = StateGraph(AgentState)
g.add_node("agent", agent_node)
g.add_node("tools", tools_node)
g.add_edge(START, "agent")
g.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
g.add_edge("tools", "agent")     # 工具执行完回到 agent，形成 ReAct 闭环
react_app = g.compile()

print(react_app.invoke({"messages": [("user", "先算 3*4，再把结果乘以 2")]}))

flowchart LR
    START([START]) --> agent[agent 节点<br/>模型思考/决策]
    agent --> cond{有 tool_calls?}
    cond -->|有| tools[tools 节点<br/>执行并回填]
    tools --> agent
    cond -->|无| END([END])

这张图就是一切 Agent 框架的内核：create_agent 只是帮你把这套「agent 节点 + tools 节点 + 条件边 + 一堆中间件」预置好。看懂手写版，再用高层 API 时就知其所以然，也知道在哪里插入自定义逻辑。

---

9. 持久化、记忆与人在回路

9.1 Checkpointer：给图装上「存档点」

编译图时传入 Checkpointer，LangGraph 会在每个超步结束自动把 State 存成一个 checkpoint（检查点），并用 thread_id（线程/会话 ID） 区分不同会话。这一套机制同时解决了三件事：短期记忆、断点续跑、时间旅行。

from langgraph.checkpoint.memory import InMemorySaver   # 内存版；生产用 SqliteSaver/PostgresSaver

checkpointer = InMemorySaver()
app = g.compile(checkpointer=checkpointer)
cfg = {"configurable": {"thread_id": "user-123"}}      # 同一 thread_id = 同一段记忆

app.invoke({"messages": [("user", "我叫小明")]}, cfg)
app.invoke({"messages": [("user", "我叫什么名字？")]}, cfg)  # 能答出“小明”，因为历史在 checkpoint 里

app.get_state(cfg)                       # 当前状态快照
list(app.get_state_history(cfg))         # 所有历史检查点 -> “时间旅行”，可回到任一节点重放

为什么用检查点而不是旧 Memory 类：检查点是对「整图状态」的通用持久化，不与某种链耦合；换个 Checkpointer 实现（内存→SQLite→Postgres/Redis）即可从单机 demo 平滑走到生产，还天然支持崩溃恢复与回放调试。

9.2 短期记忆 vs 长期记忆

flowchart LR
    subgraph 短期[短期记忆：Checkpointer 按 thread]
        t1[本次对话消息/中间状态]
    end
    subgraph 长期[长期记忆：Store 跨 thread]
        u1[用户画像/跨会话事实]
    end
    app[Agent] --> 短期
    app --> 长期

• 短期记忆：同一 thread_id 内的消息与状态，靠 Checkpointer。
• 长期记忆：跨会话、跨线程记住「用户偏好、历史结论」，用 LangGraph 的 Store（BaseStore，按 namespace 存键值）；也可外接向量库做语义检索。
• 经验上：会话上下文用 Checkpointer，需要沉淀复用的用户事实用 Store，海量私有知识用 RAG 向量库——三者分工不同。

9.3 人在回路：interrupt + Command(resume)

在节点内部调用 interrupt() 会暂停整图、把状态完整存盘，等人审批/补充后，用 Command(resume=...) 从原地精确恢复。这是「危险操作前先确认」的标准实现。

from langgraph.types import interrupt, Command

def delete_node(state):
    # 暂停并把待确认信息抛给外部；状态已持久化，进程重启也能恢复
    decision = interrupt({"action": "DELETE", "target": state["target"]})
    if decision == "approved":
        return {"result": do_delete(state["target"])}
    return {"result": "已取消"}

# 第一次运行：运行到 interrupt 处暂停，返回 __interrupt__ 信息
app.invoke({"target": "table_A"}, cfg)
# 人做决定后，带着决定从断点恢复（不会从头跑）
app.invoke(Command(resume="approved"), cfg)

设计要点（官方规则）：interrupt 不要包 try/except、不要在一个节点里调换多个 interrupt 的顺序、interrupt 之前的副作用要保证幂等（恢复时可能重放）。

---

10. 多 Agent 编排模式

当单 Agent 上下文太杂、任务需要不同专长时，拆成多个 Agent。LangGraph 官方总结了三种典型拓扑：

flowchart TB
    subgraph S[Supervisor 主管模式（最常用）]
        SU[Supervisor 调度] --> W1[研究员 Agent]
        SU --> W2[编码 Agent]
        W1 --> SU
        W2 --> SU
    end
    subgraph N[Network 网络模式]
        A1[Agent A] <--> A2[Agent B] <--> A3[Agent C]
    end
    subgraph SW[Swarm 接力模式]
        H1[前台] -->|交接控制权| H2[专家] -->|交接| H3[收尾]
    end

模式	控制权	适用
Supervisor	一个中心主管统一调度、worker 完成后回到主管	任务可拆成明确子任务、需要全局把控（最常用）
Network	Agent 间可互相调用，去中心化	角色间需要灵活协商，但易乱
Swarm / Handoff	当前 Agent 主动把控制权移交给下一个	线性流水线式客服/业务流转

Supervisor 最小骨架（每个子 Agent 本身也是一张编译好的图，作为节点接入）：

def supervisor(state):
    # 让 LLM 根据当前进展决定下一步交给哪个 worker，或结束
    decision = router_llm.invoke(state["messages"])
    return {"next": decision}          # "researcher" / "coder" / "FINISH"

g = StateGraph(State)
g.add_node("researcher", researcher_agent)
g.add_node("coder", coder_agent)
g.add_node("supervisor", supervisor)
# worker 干完都回到 supervisor，由 supervisor 条件路由到下一个 worker 或 END
g.add_conditional_edges("supervisor", lambda s: s["next"],
                        {"researcher": "researcher", "coder": "coder", "FINISH": END})
g.add_edge("researcher", "supervisor"); g.add_edge("coder", "supervisor")

经验法则：优先单 Agent + 多工具；只有当上下文隔离（子 Agent 用独立上下文避免污染）或专业分工确有收益时，才上多 Agent。多 Agent 的调试成本和 token 成本都更高。

---

11. 1.x 的统一高层 API：create_agent

手写 StateGraph 灵活但样板代码多。LangChain 1.x 用 langchain.agents.create_agent 收口了「ReAct 循环 + 工具 + 检查点 + 中间件」，它就是旧 create_react_agent 的现代版（也是 Text2SQL 笔记里用的那个）：

from langchain.agents import create_agent

agent = create_agent(
    llm,
    tools=[multiply],
    system_prompt="你是计算助手，需要计算时调用工具。",  # 旧版叫 prompt
    checkpointer=InMemorySaver(),                     # 直接获得记忆能力
    # middleware=[...]                                 # 1.x 用中间件扩展（裁剪历史、注入上下文等）
)
agent.invoke({"messages": [("user", "3*4*5 是多少")]},
             config={"configurable": {"thread_id": "t1"}})

演进关系一句话：create_agent = 第 8.5 节手写 ReAct 图 + 第 9 章持久化 + 工具调用中间件的「官方预置版」。简单场景直接用它；需要非常规循环、人在回路、多 Agent 时，退回手写 StateGraph。

---

12. LangSmith：让链路可观测、可评测

Agent 是多步、非确定的，没有可观测性几乎无法调试。LangSmith 是 LangChain 官方的配套平台（SaaS 或自托管），与上面的代码零侵入集成（设两个环境变量即可自动上报）：

• Tracing（追踪）：每次 invoke 的完整树——每个 Runnable/节点的输入输出、耗时、token、工具调用，逐层展开，是调试 Agent 的第一工具。
• Datasets / Evaluations：建评测集，对链或 Agent 做批量自动评分（含 LLM-as-judge），改 Prompt/换模型时用数据回归。
• Prompt Hub & 版本管理：Prompt 集中管理、版本化、A/B。
• Monitoring：线上监控延迟、成本、失败率。

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsxxx
export LANGSMITH_PROJECT=my-project   # 之后所有调用自动出现在 LangSmith 面板

不用 LangSmith 也能跑，但强烈建议开发期开启 trace——它把「Agent 为什么走了这条路径、哪一步工具调错」从猜谜变成可见。

---

13. LangChain vs LangGraph：选型与最佳实践

13.1 什么时候用哪个

你的场景	推荐
单次「格式化→调模型→解析」、翻译/摘要/分类	LCEL 管道
标准 RAG（检索一次再生成）	LCEL（`retriever
需要多轮工具调用、循环、自我纠错	LangGraph / create_agent
需要记忆、断点续跑、人在回路	LangGraph + Checkpointer/interrupt
多角色协作	LangGraph 多 Agent 拓扑
只是调一次模型拿结构化输出	with_structured_output，无需编排框架

13.2 最佳实践与避坑清单

1. 新项目直接用现代 API：LCEL 写链、create_agent/StateGraph 写 Agent、Checkpointer 做记忆，别碰 LLMChain/AgentExecutor/Memory 旧三件套。
2. 想清楚 State 和 Reducer：单值用覆盖、累积用 operator.add、消息用 add_messages，并行更新前先确认 reducer，否则数据被静默覆盖。
3. 任何循环都设 recursion_limit 和工具调用上限，并让模型能「明确放弃」，防止烧 token 的死循环。
4. 工具 docstring 当接口文档写：一句话说清「做什么、何时用、参数含义与格式」，这是 Agent 准确率的高杠杆点。
5. 开发期开 LangSmith trace；非确定流程务必配评测集做回归。
6. 能单 Agent 不多 Agent；多 Agent 的收益主要在上下文隔离与专业分工，不要为了架构而架构。
7. 模型与框架解耦：业务逻辑面向 langchain-core 抽象写，换厂商只改 ChatXxx 初始化（本笔记代码从 DeepSeek 换到 OpenAI 只需改 base_url/model/key）。
8. 推理型模型注意多轮回传：像 deepseek-v4-flash 这类带思维链的模型，老的 legacy AgentExecutor 可能因 reasoning 字段回传问题报错，现代 LangGraph/create_agent 已处理——这也是弃用旧 API 的现实原因之一。

---

14. 参考资料

1. LangChain 官方文档（1.x，含 LCEL / 工具 / 检索）：https://docs.langchain.com/oss/python/langchain/overview
2. LangGraph 官方文档（StateGraph / 持久化 / 人在回路 / 多 Agent）：https://docs.langchain.com/oss/python/langgraph/
3. LangGraph 概念——Low-level Concepts、Persistence、Interrupts、Multi-agent：https://langchain-ai.github.io/langgraph/
4. LCEL 教程（Why use LCEL / How to chain runnables）：https://python.langchain.com/docs/how_to/output_parser/
5. create_agent API 参考（1.x 统一 Agent API）：https://docs.langchain.com/oss/python/langchain/agents
6. LangSmith 文档（Tracing / Evaluation）：https://docs.smith.langchain.com/
7. ReAct 论文（手写 Agent 循环的理论来源）：https://arxiv.org/abs/2210.03629
8. LangChain 官方 GitHub 与示例库：https://github.com/langchain-ai/langchain 、https://github.com/langchain-ai/langgraph

---

配套动手材料（同目录）：langchain_langgraph_demo.ipynb（Model I/O → LCEL 原语 → 工具调用 → 手写 StateGraph/ReAct → Checkpointer 记忆 → interrupt 人在回路 → create_agent，已用 deepseek-v4-flash 逐格实测）、requirements.txt、.env.example。