书架 · Python 学习系列 · 09 · LangChain + LangGraph：从学习到工程化返回目录 →

09 · LangChain + LangGraph：从学习到工程化

本章目标：在你已手写过 100 行裸 Agent（第 05 章）的基础上，系统掌握 LangChain 1.0 + LangGraph 1.0——从三个基础原语，到图编排、持久化、人工审批、流式、结构化输出、中间件，最后到部署与观测，完成从「会调 API」到「能交付生产级 Agent 服务」的跨越。

前置：04-LLM应用开发、05-Agent开发（务必先手写过 100 行版）、07-Web服务-FastAPI（部署层会用到）。

📌 本章所有 API 形态均核对自官方文档（docs.langchain.com）与官方 1.0 发布博客，访问日期 2026-09-05，出处见文末。

0. 本章怎么读：七层地图

每一层都按同一个节奏讲：是什么 → 为什么 → 怎么用（对照你手写过的代码）。你不需要每层都精通，但每层的"为什么"都值得理解——它们决定了你在真实项目里怎么选型。

1. L0 全景：生态四件套与定位

1.1 是什么：四个组件各管一段

为什么是这个分层：官方在 1.0 发布博客里明确了两点——① LangChain 的 agents 构建在 LangGraph 之上，从高层 API 起步不会锁死（需要时下沉）；② 老的 AgentExecutor 退役，create_agent 成为标准入口。也就是说：高层是底层的糖，底层永远是图。这条事实链决定了本章的讲法：先高层（L1），再底层（L3），最后生产件（L4/L5）。

1.2 为什么需要框架：你手写的 100 行缺什么

第 05 章的 100 行 Agent 能跑通循环，但放到生产环境，下面每一条都要自己造：

⭐ 判断标准（贯穿本章）：如果上面这张表里你一条都不需要——脚本项目，继续手写，更快更好调试。需要其中 2 条以上，框架开始回本。

1.3 安装

uv add langchain langgraph          # 高层 + 图运行时
# 按需补充：
uv add langchain-openai             # OpenAI 及一切兼容端点（含 DeepSeek/Qwen/GLM）
uv add "langgraph-checkpoint-sqlite"  # SQLite 持久化

官方同时维护各 provider 的集成包（以及大量社区集成），完整清单见官方 models/integrations 文档（文末出处）。国产三家在第 04 章已核实过 OpenAI 兼容参数，本章示例统一用 ChatOpenAI(base_url=...) 方式接入，换 model 与 base_url 即可。

2. L1 LangChain 基础层：三个原语 + 一个入口

2.1 init_chat_model：模型访问的统一入口

是什么：一个工厂函数，传 "provider:model" 字符串（或先装好的模型对象），返回统一的 ChatModel 接口对象。

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", temperature=0)      # 字符串形式
model = init_chat_model("anthropic:claude-sonnet-4-6")        # 换家只改字符串
# 国产 / 自建兼容端点（第 04 章核实过的参数）：
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
    model="deepseek-v4-pro",
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

为什么：所有 provider 的 ChatModel 实现同一接口（invoke / stream / bind_tools / ainvoke）——你的业务代码不 import 任何具体厂商类，换模型 = 改一个字符串。这就是 Java 里"面向接口编程 + 工厂"在 LLM 层的翻版。官方还支持运行时切换（config={"configurable": {"model": ...}}），同一个服务按请求路由不同模型。

怎么用（工具绑定）：bind_tools 把工具挂到模型上——它就是你手写的 tools=[...] 参数 + schema 生成的合体：

from pydantic import BaseModel, Field

class GetWeather(BaseModel):
    """查询指定城市当前天气"""
    city: str = Field(description="城市名")

model_with_tools = model.bind_tools([GetWeather])     # 传 Pydantic 类或 @tool 函数
resp = model_with_tools.invoke("成都天气怎么样")
resp.tool_calls      # [{'name': 'GetWeather', 'args': {'city': '成都'}, ...}]

2.2 消息类型：从裸 dict 到结构化对象

是什么：langchain.messages 提供与 OpenAI 协议同构的消息类——SystemMessage / HumanMessage / AIMessage / ToolMessage。

为什么：手写版里 messages 是裸 dict，取工具调用要 msg["tool_calls"] 且各家字段有细微差异；统一对象抹平差异（AIMessage.tool_calls 已解析成 dict、ToolMessage(tool_call_id=...) 强制回带 id），并给 LangSmith 追踪提供了标准结构。输入端仍兼容裸 dict（{"role": "user", "content": ...}），迁移零成本。

from langchain.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

messages = [
    SystemMessage("你是会用工具的助手"),
    HumanMessage("成都天气怎么样？"),
    # AIMessage(tool_calls=[...]) → 工具执行后：
    ToolMessage(content="26℃，多云", tool_call_id="call_a1"),
]

2.3 @tool：工具定义的标准姿势

是什么：装饰器。从函数的类型注解 + docstring 自动生成 tools schema 并包装成可传给 agent 的工具对象。

为什么：你手写的 @tool 注册表（inspect 签名 + 手拼 schema）约 25 行且只支持 string 参数；官方 @tool 生成完整 JSON Schema（int/float/枚举/嵌套模型都行），description 直接取 docstring——模型看得更准，选工具就更准。

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市当前天气。当用户问到任何城市的实时天气时使用。"""
    return "26℃，多云"          # ↑ docstring = 模型读的工具说明（第 05 章坑 3）

@tool(return_direct=True)       # 工具结果直接作为最终答案，跳过最后一次 LLM 调用（省钱）
def get_order_status(order_id: str) -> str:
    """查询订单状态"""
    return f"订单 {order_id} 已发货"

2.4 create_agent：你 100 行的"豪华包装版"

是什么：1.0 的标准 Agent 入口。传入模型、工具、系统提示词，返回一个可 invoke / stream 的 Agent 对象——内部是一张 LangGraph 图（agent 节点 ↔ tools 节点的循环）。

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"{city}：26℃，多云"

agent = create_agent(
    model="openai:gpt-5.5",             # 字符串或 ChatModel 对象
    tools=[get_weather],
    system_prompt="你是简洁的助手",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "成都天气怎么样？"}]}
)
print(result["messages"][-1].content_blocks)    # 最终答案（官方 quickstart 的取值方式）

对照表（务必逐行体会）：

四行核心代码对一百行，被藏起来的正是 L4 那些"生产件"的实现——这就是框架的真实价值所在，也是你要逐层往下学的原因。

3. L2 Runnable / LCEL：统一接口与管道组合

3.1 是什么

Runnable 是 LangChain 的一切可执行对象的统一接口：invoke / stream / batch / ainvoke（+ astream）。模型、工具、prompt 模板、解析器、Agent、LangGraph 图——全是 Runnable。LCEL（LangChain Expression Language）是基于它的组合语法：| 管道、RunnableParallel 并行等。

from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "把用户输入翻译成{language}"),
    ("user", "{text}"),
])
chain = prompt | model | StrOutputParser()      # ← LCEL 管道：每个元素都是 Runnable

chain.invoke({"language": "英文", "text": "你好世界"})
for token in chain.stream({"language": "英文", "text": "你好世界"}):   # 流式贯穿整条链
    print(token, end="", flush=True)

3.2 为什么

1. 流式天然贯穿：chain.stream() 从模型 token 级流到输出端，中间环节自动做增量传递——对比手写版要自己拼接 chunk（第 04 章）
2. 组合正交：任何 Runnable 可接入任何位置，并行/回退/重试都有标准件（RunnableParallel、RunnableWithFallbacks）
3. 与 LangGraph 同构：图的每个节点就是 Runnable；create_agent 内部就是"模型节点 + 工具节点"的图。学会 Runnable，读框架源码就通了

from langchain.schema.runnables import RunnableParallel, RunnablePassthrough

# 并行分支 + 汇聚（≈ CompletableFuture.allOf）
setup = RunnableParallel(
    context=retriever,                # 取参考资料
    question=RunnablePassthrough(),   # 原样透传
)
rag_chain = setup | prompt | model | StrOutputParser()

3.3 何时用 LCEL

1.0 之后官方的推荐路径是：Agent 场景直接 create_agent（内部已是图）；LCEL 最适合无循环的静态流水线（翻译链、RAG 前置处理、批量分类）。有"模型决定下一步"的循环需求时，不要再拿 LCEL 硬拼（旧时代 AgentExecutor 式的拼法已退役），直接上 L3 的图。

4. L3 LangGraph 核心：把 while 循环显式化为图

4.1 为什么需要图（本章最重要的一节）

手写 run_agent() 的 while 循环有三个结构性缺陷：

1. 状态隐形：循环状态（messages）只存在于函数局部变量里——进程一死全丢，无法恢复、无法审计
2. 控制流硬编码：循环里塞不进"人工审批""并行分支""按状态跳转"这类需求，只能 if 叠 if
3. 没有"步"的概念：出了问题不知道执行到哪一步，无法回放

LangGraph 的答案：把循环显式建模为状态图（StateGraph）——

• 状态（State）显式声明：一个 TypedDict，不再是隐形的局部变量
• 节点（Node）= 一步：普通函数，接收 state、返回 state 更新
• 边（Edge）= 控制流：固定边 + 条件边（函数决定下一跳）→ 循环、分支都是一等公民
• 每步可存档（checkpointer，L4）：从此可恢复、可回放、可中断

🧠 心智迁移：Java 老兵可以把它理解为带持久化状态的有限状态机（FSM/Spring StateMachine），只不过转移函数里跑的是 LLM。

4.2 State 与 reducer：状态怎么合并

是什么：状态用 TypedDict 声明；Annotated[list, add] 这类注解声明合并策略（reducer）——节点返回的值如何并入现有状态。

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str                              # 默认策略：覆盖
    bar: Annotated[list[str], add]        # reducer=add：追加合并（concat）

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}     # 节点只返回“增量”

# 连续执行 node_a、node_b（都返回 bar=["x"]）后：bar=[...] 是累加，foo 是最后写入

为什么：Agent 循环的本质是"消息列表不断追加"——MessagesState 内置状态就是 {"messages": Annotated[list, add_messages]}，add_messages reducer 负责：追加新消息、同 id 消息覆盖更新。你手写版里 messages.append(...) 的每一处，在图里都变成"节点返回增量，reducer 帮你合"。

4.3 节点与边：最小图

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraph, START, END

llm = init_chat_model("openai:gpt-5.5")

def node(state: MessagesState):
    new_message = llm.invoke(state["messages"])   # 读状态 → 调模型
    return {"messages": [new_message]}            # 返回增量（reducer 自动追加）

builder = StateGraph(MessagesState)
builder.add_node(node)                    # 函数名即节点名（也可 add_node("名字", fn)）
builder.add_edge(START, "node")           # 入口边
builder.add_edge("node", END)             # 到 END 结束
graph = builder.compile()                 # 编译成可执行图（结构校验在此发生）

result = graph.invoke({"messages": [{"role": "user", "content": "Hello"}]})

条件边让控制流"活"起来（这是 Agent 循环的关键）：

from langgraph.graph import END, MessagesState, StateGraph

builder = StateGraph(MessagesState)         # 承接 4.3 的最小图

def should_continue(state):                 # 路由函数：返回下一节点名
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"                    # 模型要调工具 → 去 tools 节点
    return END                            # 没有了 → 结束

builder.add_conditional_edges("agent", should_continue)   # agent 之后由它决定去向

4.4 完整示例：手写 ReAct 图（第 05 章 100 行的 LangGraph 版）

"""react_graph.py —— LangGraph 版 ReAct Agent
依赖: uv add langchain langgraph langchain-openai
"""
import os
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode

llm = ChatOpenAI(                                     # 国产兼容端点（第 04 章核实）
    model="deepseek-v4-pro",
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

@tool
def get_weather(city: str) -> str:
    """查询指定城市当前天气"""
    fake = {"成都": "26℃，多云", "北京": "15℃，晴"}
    return fake.get(city, f"{city}：暂无数据")

tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)                # ① 挂工具

def agent(state: MessagesState):                      # ② agent 节点：调模型
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def should_continue(state: MessagesState):            # ③ 条件边：循环的本体
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))            # ToolNode：执行+回填一步到位
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")                    # 循环边：tools → agent
graph = builder.compile()

result = graph.invoke(
    {"messages": [{"role": "user", "content": "成都和北京哪个更热？"}]},
    config={"recursion_limit": 16},                   # 循环上限（你的 max_rounds）
)
print(result["messages"][-1].content)

对照手写版：while 循环消失了，变成了 tools → agent 的循环边；"执行工具 + 回填"整段被 ToolNode 吃掉。每行代码都能在第 05 章找到原型——这就是"先手写再学框架"的复利。

4.5 图执行模拟器（交互动效）

下面把 4.4 这张图跑给你看（含一个危险工具触发 interrupt() 人工审批的完整过程，对应 L4 第 2 节）。点「下一步」观察节点高亮、右侧解说与底部的检查点时间轴：

💡 底部时间轴就是 checkpointer 的可视化：每个节点跑完自动存一档。没有它，interrupt() 之后的"恢复"无从谈起——这就是 L4 要讲的内容。

4.6 预制件与 Functional API（知道即可）

• langgraph.prebuilt.create_react_agent：把 4.4 的图直接给你（模型+工具一步成图）；LangChain 1.0 的 create_agent 是它的上层强化版（带 middleware 生态、结构化输出等），新项目从 create_agent 起步
• Functional API（@task / @entrypoint）：官方提供的"代码即图"写法——不显式 add_node，用装饰器组合，同样享受 checkpointer 与 interrupt。适合"图结构不复杂但想保留代码可读性"的场景（官方 choosing-apis 页有完整对比）：

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def step_1(query): ...

@entrypoint(checkpointer=InMemorySaver())     # 官方 use-functional-api 模式
def graph(input_query):
    result_1 = step_1(input_query).result()
    result_2 = step_2(result_1).result()      # task 之间自动成为图节点
    return result_2

5. L4 工程化生产件（框架的真正价值）

5.1 Checkpointer：持久化与多轮记忆

是什么：给图挂一个存档器，每个节点执行完自动保存一份完整状态快照（checkpoint），按 thread_id 组织。

from langgraph.checkpoint.memory import InMemorySaver          # 原型用（重启即失）
# from langgraph.checkpoint.sqlite import SqliteSaver          # 生产单机用 SQLite
# from langgraph.checkpoint.postgres import PostgresSaver      # 生产集群用 PG

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

config = {"configurable": {"thread_id": "user-42-session-7"}}   # 会话 = thread
graph.invoke({"messages": [{"role": "user", "content": "我叫 macao"}]}, config)
# ……进程重启后，同 thread_id 继续聊，模型还记得你：
graph.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config)
# → "你叫 macao"（历史从 checkpoint 恢复，不用你自己存 messages！）

为什么：① 多轮记忆免费获得——手写版里你要自己存 SQLite（第 05 章第 3.5 件事）；② interrupt 的前提（见下）；③ 时间旅行：graph.get_state_history(config) 拿到全部快照，可回放到任意一步调试。

5.2 interrupt + Command：人工审批（HITL）

是什么：节点内调用 interrupt(payload) 会立刻冻结图的执行（状态存进 checkpoint），把 payload 抛给外部；人工决策后用 Command(resume=值) 从断点继续，interrupt() 的返回值就是 resume 传入的值。

from langgraph.types import Command, interrupt

CONFIRM_TOOLS = {"delete_order", "send_email"}

def agent(state: MessagesState):
    last = llm_with_tools.invoke(state["messages"])
    return {"messages": [last]}

def human_approval(state: MessagesState):
    """危险工具执行前的审批节点"""
    pending = state["messages"][-1].tool_calls
    dangerous = [t for t in pending if t["name"] in CONFIRM_TOOLS]
    if not dangerous:
        return Command(goto="tools")                 # 无危险 → 直接过
    answer = interrupt({                             # 有危险 → 冻结，等人工
        "question": "以下操作需要审批",
        "tool_calls": dangerous,
    })
    if answer is True:
        return Command(goto="tools")                 # 批准 → 执行
    return Command(goto="agent")                     # 拒绝 → 回 agent 让它改道

# ……外部（比如 FastAPI 路由）拿到 interrupt 后，收集人工输入再恢复：
graph.invoke(Command(resume=True), config)           # 官方 interrupts 页标准模式

为什么：第 05 章"生产底线"第 5 条（human-in-the-loop）在 Web 服务里手写非常痛苦——HTTP 请求是无状态的，"弹窗等人点确认"意味着要把整个执行现场存下来再恢复。checkpointer + interrupt 把这件事变成了两行代码：冻结即存档，恢复即续跑。这是手写版最难补、框架最值钱的一块。

5.3 流式：多模式事件流

是什么：Agent/图的 stream() 支持多种 stream_mode，一次拿到不同粒度的事件（官方 streaming 页形态）：

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"{city}：26℃，多云"

agent = create_agent(model="openai:gpt-5.5", tools=[get_weather])   # 接 2.4

input_message = {"role": "user", "content": "成都天气怎么样？"}
for chunk in agent.stream(
    {"messages": [input_message]},
    stream_mode=["messages", "updates", "custom"],   # 官方示例的组合
    version="v2",                                    # 事件协议版本
):
    if chunk["type"] == "messages":                  # ① token 级：打字机效果
        token, metadata = chunk["data"]
    elif chunk["type"] == "updates":                 # ② 节点级：谁完成了什么
        for source, update in chunk["data"].items():
            pass                                     # source ∈ {model, tools}
    elif chunk["type"] == "custom":                  # ③ 自定义事件（middleware 发的）
        pass

为什么：前端 AI 应用的标配体验——"正在查天气…"（updates）→ 工具结果卡片（custom）→ 逐字回答（messages）。LangGraph 侧还有 stream_events(version="v3") 的更细粒度事件 API（含子图事件，官方 functional-api 页有完整示例）。接到第 07 章的 FastAPI SSE 接口上，事件直接透传即可。

5.4 结构化输出：ToolStrategy / ProviderStrategy

是什么：create_agent 直接吃 response_format，跑完循环后给你一个类型安全的结构化结果（官方 structured-output 页）：

from typing import Literal
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy

class ProductReview(BaseModel):
    """Analysis of a product review."""
    rating: int | None = Field(description="1-5 分", ge=1, le=5)
    sentiment: Literal["positive", "negative"]
    key_points: list[str] = Field(description="要点，小写，每条 1-3 词")

agent = create_agent(
    model="openai:gpt-5.5",
    tools=tools,                              # Agent 照常用工具收集信息
    response_format=ToolStrategy(ProductReview),
)
result = agent.invoke({"messages": [
    {"role": "user", "content": "分析评论：'非常好，5 星，发货快就是贵'"}
]})
result["structured_response"]
# ProductReview(rating=5, sentiment='positive', key_points=['fast shipping', 'expensive'])

两种策略（官方定义）：ToolStrategy 用"隐形工具调用"拿结构（兼容面广）；ProviderStrategy 优先用 provider 原生 JSON Schema 能力，不支持时回退工具调用。为什么：第 04 章你手写 JSON mode + Pydantic 校验 + 失败重试约 30 行；这里 Agent 循环和结构化一并解决，且校验失败框架内重试。

5.5 Middleware：请求/工具执行的钩子层

是什么：create_agent(..., middleware=[...])——在模型调用前后、工具执行前后插自定义逻辑的正规位置（1.0 新增，官方 middleware 文档）：

from langchain.agents.middleware import AgentMiddleware

class ToolActivityMiddleware(AgentMiddleware):
    transformers = (ToolActivityTransformer,)    # 官方示例：注册自定义流事件变换器

agent = create_agent(model="openai:gpt-5.5", tools=tools,
                     middleware=[ToolActivityMiddleware()])

为什么：安全护栏（拦危险 prompt）、审计日志、工具结果兜底改写、给流加自定义事件——这些"横切关注点"在 Java 里你用 AOP/Filter，在这里就是 middleware。没有它你就得侵入节点函数，或者 fork 框架代码。

5.6 子代理（Subagents）：工具即代理

是什么：把"启动一个子 Agent"封装成主 Agent 的工具，子 Agent 结果通过 Command(update=...) 回写主图状态（官方 multi-agent/subagents 页模式）：

from typing import Annotated
from langchain.tools import InjectedToolCallId, tool
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool("research_subagent", description="让子代理做深度调研并返回结论")
def call_subagent(query: str, tool_call_id: Annotated[str, InjectedToolCallId]) -> Command:
    result = subagent.invoke({"messages": [{"role": "user", "content": query}]})
    return Command(update={
        "messages": [ToolMessage(content=result["messages"][-1].content,
                                 tool_call_id=tool_call_id)],
    })

为什么：主代理保持轻上下文（只看结论），重活丢给子代理（自带独立上下文）——这是对抗"上下文爆窗"（第 05 章第 3.4 件事）的主流架构，你在 ZCode 里让我派 Explore 子代理干活就是同一个模式。

5.7 生产清单

6. L5 部署与观测

6.1 LangSmith：没有 trace 的 Agent 调试是盲调

是什么：LangChain 官方观测平台。设置环境变量（LANGSMITH_API_KEY + 开启 tracing，具体以官方文档为准）后，每次 invoke 自动上报：每层 prompt、每次工具调用入参出参、耗时、token 用量、错误——树状 trace 时间线。

为什么：Agent 的行为 = 模型决策 + 你的代码 + 数据，三者交织。出了"它为什么调了这个工具"的问题，日志只能看到结果，trace 能看到每一步的输入输出。这相当于给 LLM 应用上了 SkyWalking + Arthas 的合体。对第 05 章手写版，等价物是你在 run_agent 里手打的 print——框架版免费拿全家桶。

6.2 部署形态三档

1. langgraph dev（本地开发）：官方 CLI 起本地服务 + LangGraph Studio（可视化调试界面：看图、看 checkpoint、时间旅行回放）——学习期最有用的工具
2. 自托管 LangGraph Server（Docker）：把图变成正式 API 服务（REST + 任务队列 + 流式接口），长任务、interrupt 等待审批的挂起状态由服务端管理；自己 k8s/容器化部署
3. LangGraph Platform（托管）：官方云托管上面这一切

与第 07 章的取舍：create_agent 是普通 Python 对象，直接嵌进 FastAPI 路由里就能跑（很多项目这样就够了）；当你需要"跨请求的 interrupt 等待、长时间任务队列、多实例共享 checkpoint"时，才值得把图抽到独立的 LangGraph Server，FastAPI 退化为网关。

6.3 上线检查单

• [ ] checkpointer 选型落实（InMemory→SQLite→Postgres），thread_id 设计 = 业务会话 id
• [ ] 危险工具全部进 CONFIRM_TOOLS + interrupt 审批
• [ ] recursion_limit、模型 timeout/max_retries、密钥环境变量化
• [ ] LangSmith trace 开启 + 告警（错误率/成本突增）
• [ ] SSE 透传链路验证（stream_mode events → FastAPI → 前端）

7. L6 决策树与学习路径

7.1 选型决策树

你的需求？
├─ 脚本/单次任务，无持久化无审批 ──────────→ 手写（第 05 章 100 行），最快最透明
├─ 标准 Agent：工具 + 循环 + 流式 ──────────→ LangChain create_agent（L1）
│    └─ 还要结构化输出/中间件/子代理 ───────→ create_agent + response_format/middleware（L4）
├─ 非标准控制流：审批节点、并行分支、多 Agent 协作 → LangGraph StateGraph（L3）
│    └─ 图结构简单、想保持代码可读 ─────────→ Functional API @task/@entrypoint
└─ 长任务/队列/多实例/需要 Studio 调试 ─────→ LangGraph Server/Platform（L5）

官方在 choosing-apis 页给出的同款建议：prebuilt（高层）起步 → Graph API 要控制力 → Functional API 要可读性。永远从能满足需求的最高层开始，锁死顾虑在 1.0 后已消除（高层就构建在低层上，随时下沉）。

7.2 两周路径（在已学完 01–05/07 章基础上）

7.3 自测清单

• [ ] 四件套各管什么？LangChain 的 agents 构建在什么之上（1.0 事实）？
• [ ] create_agent 藏掉了你 100 行里的哪六样？
• [ ] reducer 是什么？MessagesState 的 messages 为什么是追加不是覆盖？
• [ ] while 循环在图里变成了哪两个东西？
• [ ] checkpointer 与 interrupt 为什么是绑在一起的？
• [ ] ToolStrategy 和 ProviderStrategy 的区别？
• [ ] 什么时候该从 create_agent 下沉到 StateGraph？

7.4 练习

1. 用 create_agent 复刻第 05 章的天气+算术 Agent，对比代码量与可调试性（LangSmith trace）。
2. 给 4.4 的 ReAct 图挂 SqliteSaver，杀掉进程重启后同 thread_id 继续对话，验证记忆恢复。
3. 加 delete_file 危险工具 + 审批节点：拒绝时应走回 agent 改道而不是报错。
4. 把练习 3 的图接到 FastAPI：/chat（普通）+ /chat/stream（SSE 透传 stream_mode 事件）+ /approve（接收审批结果后 Command(resume=...)）。

参考与出处

以下全部为官方一手来源（访问日期：2026-09-05）：

⬅️ 返回目录 ｜ 🔥 ReAct 深剖（交互页） ｜ ➡️ 08-实战项目与资源