LangGraph配合LCEL构建复杂Agent

可视化演示

展示LangGraph状态机多步流程编排、LCEL声明式链路单步执行、两者组合的状态传递与人机交互节点。切换模式查看Agent不同构建机制。

状态机原理

后厨的墙上挂着一块巨大的排菜板，每一道菜是一张卡片，卡片上记录着“配菜 + 主料 + 辣度 + 特殊备注”。排菜总管只看这块板，就知道哪道菜在备料、哪道在掌勺、哪道待质检。LangGraph 就是这块排菜板背后的调度系统——它把整个工作流画成一张图来驱动。

图由两类元素构成：节点（node）和边（edge）。节点是“工作台”，执行具体的处理函数；边是“传菜路线”，决定 A 台处理完后去 B 台还是 C 台。整张图运行时会维护一份所有节点共享的状态（state）：每个节点读取状态、加工、再写回，就像每个台位在订单卡上做批注。

在代码里，先用 StateGraph(StateSchema) 声明状态的“数据结构”，再用 add_node("备料", func) 注册工作台，用 add_edge / add_conditional_edge 定义路线，最后 compile() 把图“点火”成可执行对象。LangGraph 的核心循环就是：当前状态 S 沿着边找到下一节点，执行节点函数，更新状态，再找下一条边。

有了这张“排菜板”，再复杂的流程都被拆成了清晰的台位与传菜路线。但请注意：总管只负责“谁在哪儿做”，至于每个台位具体怎么把菜做好，那是菜谱卡的事——这就是接下来要讲的 LCEL。

LCEL声明式链路

每位厨师台前都贴着一张菜谱卡：“洗净 → 切丝 → 大火爆炒 → 出锅装盘”。它用一条直线把工序串起来，厨师不用想下一步干嘛，照着卡走就行。LCEL（LangChain Expression Language）就是这张菜谱卡——它把“提示词整理、模型调用、结果解析”串成一条声明式的流水线。

LCEL 的核心是竖线管道运算符 |：左边产出的东西自动变成右边的输入，像传送带一样一环扣一环。任何实现了 Runnable 接口的对象——提示模板、大模型、输出解析器——都可以被 | 串起来。这就是“可运行对象”（Runnable）的统一契约：有输入、有输出、能调用。

一条典型的链长这样：prompt_template | llm | output_parser。它最终是一个 RunnableSequence，支持 invoke（同步执行）、stream（流式输出）、batch（批量处理）。声明式的妙处在于：代码读起来就是流程本身，不需要手写任何胶水逻辑。

菜谱卡让每个台位的“单步动作”标准化、可复用。但光有菜谱卡，餐厅还是转不起来——必须由排菜总管把卡发到对应工作台，并按订单顺序调度。下一节，我们就正式把 LCEL 链“递”给 LangGraph 的每个节点。

组合集成方式

排菜总管把菜谱卡钉在每个工作台正上方：“备料台请按《凉菜卡01》执行，掌勺台请按《热菜卡07》执行”。工作台换了人，流程却不变——因为“台位职责”和“具体怎么做”被拆开了。LangGraph 与 LCEL 的组合，拆的正是这两件事。

组合的方法非常简单：把一个 LCEL 链“包”进一个普通函数，这个函数接收状态、调用链、返回增量字段，再把这个函数注册成 LangGraph 的节点。LangGraph 根本不关心节点内部是什么，它只认“输入状态 → 输出状态”这个契约。

典型的节点函数长这样：

然后 graph.add_node("node_name", node) 把它挂上图，定义好边，最后 graph.compile()。运行时，LangGraph 负责挑节点、传状态、收结果，LCEL 负责在节点内部完成“提示 → 生成 → 解析”的精细活。

菜谱卡钉上了墙，总管一声令下，台位就按卡开工。但这里有个关键问题：台位之间怎么把订单信息完整传下去？辣度、忌口、加急备注，如何在十几个台位间准确流转？这就是状态传递机制的活。

def node(state):    result = chain.invoke({"question": state["question"]})    return {"answer": result}

状态传递机制

订单卡从进厨房的第一秒起，就在被各个台位批注：备料台写上“鱼已切好”，掌勺台写上“辣度减半已执行”，装盘台写上“少放葱”。没人重抄整张卡，每个人只在自己负责的格子里写几笔——但任何台位拿到卡，都能看到完整信息。这就是 LangGraph 状态传递的本质。

状态的结构由 StateSchema（通常是一个 TypedDict）提前定义好，例如 {"question": str, "draft": str, "messages": list}。每个节点执行完只返回自己负责的那部分字段，即增量；LangGraph 会用“归约器”（reducer）把增量合并回总状态，节点之间永远共享同一份最新快照。

对于普通字段，默认的归约规则是“覆盖”：新值替换旧值。对于需要累积的字段（比如多轮对话记录），要显式声明归约器，最常用的是 add_messages——它把新消息追加到消息列表末尾，而不是覆盖。这样状态既全局可见，又避免了每个节点重复携带全部数据。

订单卡就这样“越写越厚”地穿过整条后厨流水线——这正是复杂 Agent 的血液系统。有了血液（状态），还得有大脑（编排逻辑）：什么时候该走哪条路、要不要调用外部工具，这就是 Agent 的编排能力。

Agent编排能力

有经验的排菜总管不会傻走一条线。看到订单备注“客人对花生过敏”，他立刻把菜从“常规炒台”改派到“无过敏原专台”；发现冰箱缺了鲈鱼，他马上派小工去市场采购。复杂 Agent 也是这样：根据当前状态，动态决定下一步走哪条边。

LangGraph 提供了条件边（conditional edge）：一个路由函数接收状态，返回“下一个节点的名字”。这把 if-else 从代码里搬进了流程图——流程走向由状态数据驱动，而不是写死的。这是 Agent 区别于“固定流水线”的核心能力。

再加上大模型本身的推理能力，一个 Agent 循环就成型了：模型判断“需要查资料” → 路由到工具节点 → 工具返回结果写回状态 → 模型再判断“可以回答了” → 路由到输出节点。多步推理（multi-step reasoning）和工具调用（tool calling）就是这样被编排成一张动态图。

有了会看单改路线的总管，餐厅能接住各种突发状况。但有一类情况连总管都不敢拍板——比如贵宾宴席的主菜，他必须请主厨亲自尝一口再决定上不上。这种“流程停下来等人”的机制，就是人机交互节点。

人工介入控制

国宴级别的订单，排菜总管会设定一个“主厨试菜点”：菜做完先不上桌，端到主厨面前尝一口，要么盖章放行，要么打回重做。在这个点上，整条流水线是暂停的——所有人等着主厨的裁决。这就是 human-in-the-loop（人在回路）。

LangGraph 提供了中断机制：在某个节点执行前（interrupt_before）或执行后（interrupt_after）设置断点。图运行到断点时，会保存当前状态并挂起，等待外部（人类）输入，之后从断点继续，而不是从头执行。

配合持久化（checkpointer）功能，中断后的状态可以安全保存和恢复。常见用法：Agent 生成草稿 → 中断 → 人审阅并修改 → 恢复执行后续节点。这个能力把 Agent 从“全自动工具”升级为“可控的生产流程”——就像主厨的章，让每一道大菜都有人把关。

主厨尝完菜，流程继续。但人不可能永远盯着机器——如果某个台位总是把菜做坏，总管就得定规矩：最多重做几次、重做失败怎么办。这就是容错重试策略。

容错重试策略

掌勺台偶尔会糊锅。总管定的规矩是：糊了，重新炒一次；又糊，再重炒一次；连炒三次还糊，立刻上报主厨换菜，绝不让客人干等。谁都不想上菜出错，但一家成熟的餐厅，靠的不是“永不犯错”，而是“错了知道怎么补救”。

复杂 Agent 的节点里，大模型调用、工具调用都可能抛出异常：网络抖动、API 限流、返回格式不对。所以每个关键节点要做两层防护：一是捕获异常（try / except），二是按策略重试（retry）。

重试策略要控制三个参数：最大重试次数 N_max、退避时长（如指数退避 1s → 2s → 4s）、重试条件（只在可恢复异常时重试）。超过 N_max 就进入兜底分支（Fallback）：返回预设文案、降级模型，或转人工。LangGraph 把“重试逻辑”封装在节点内部，配合外层编排实现更精细的重试路由。

到这里，后厨的整套管理体系就齐了：排菜板管全流程、菜谱卡管单步、订单卡传递状态、总管动态派单、主厨试菜把关、糊锅重炒兜底。这就是 LangGraph 与 LCEL 组合构建复杂 Agent 的完整图景——一张自带秩序的生产流水线。

AI 应用场景

Python 演示

from typing import TypedDict from langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAIfrom langgraph.graph import END, StateGraph  class AgentState(TypedDict):    question: str    draft: str    answer: str  llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) summarize_chain = (    ChatPromptTemplate.from_template("请先用一句话提炼问题的核心：{question}")    | llm    | StrOutputParser()) answer_chain = (    ChatPromptTemplate.from_template(        "请结合要点【{draft}】完整回答用户问题：{question}"    )    | llm    | StrOutputParser())  def node_summarize(state: AgentState) -> dict:    return {"draft": summarize_chain.invoke({"question": state["question"]})}  def node_answer(state: AgentState) -> dict:    return {        "answer": answer_chain.invoke(            {"question": state["question"], "draft": state["draft"]}        )    }  graph = StateGraph(AgentState)graph.add_node("summarize", node_summarize)graph.add_node("answer", node_answer)graph.set_entry_point("summarize")graph.add_edge("summarize", "answer")graph.add_edge("answer", END) app = graph.compile() if __name__ == "__main__":    result = app.invoke({"question": "LangGraph和LCEL如何搭配使用?"})    print(result["answer"])

1. class AgentState(TypedDict):用类型字典定义状态的键与类型，即“订单卡模板”。
2. summarize_chain = (LCEL 竖线管道把提示模板、模型、解析器串成一条链。
3. | llm管道运算符把上一步输出自动送进大模型。
4. def node_summarize(state: AgentState) -> dict:节点函数接收状态，是 LangGraph 与 LCEL 之间的适配器。
5. graph = StateGraph(AgentState)用状态结构创建状态机图。
6. graph.add_node("summarize", node_summarize)把 LCEL 包装节点注册到图上。
7. graph.set_entry_point("summarize")指定流程入口节点。
8. app = graph.compile()编译为可执行应用。
9. result = app.invoke({"question": "LangGraph和LCEL如何搭配使用?"})喂入初始状态，启动整条流水线。

常见误区

面试题