短剧《拥有纯阳仙体，我无敌了》剧情介绍
2026-07-23 3419969
2026-07-23 0
在接触 LangGraph 之前，大多数开发者已经熟悉了 LangChain。LangChain 通过 LCEL（LangChain Expression Language）提供了一种极其简洁的方式来串联 LLM 调用：
# LangChain 的经典写法：线性流水线chain = prompt_template | model | output_parserresult = chain.invoke({"topic": "人工智能"})
这种写法优雅、直观，但它隐含了一个前提：流程是固定的、单向的、无环的。数据从起点流向终点，不会回头，也不会在中途分叉后再汇合。
然而，现实世界的 AI 应用远比这复杂：
这些场景需要的不是一个线性管道，而是一个有状态、可循环、可中断的执行引擎——这正是 LangGraph 诞生的原因。
一句话总结：LangChain 让你快速跑起来，LangGraph 让你跑得复杂、跑得可控、跑得持久。两者不是替代关系，而是互补关系——你用 LangChain 的组件（模型、工具、检索器）作为积木，用 LangGraph 作为把这些积木粘合在一起的“蓝图与施工队”。
如果你遇到了以下任何一个问题，就该考虑从 LangChain 升级到 LangGraph：
LangGraph 的所有能力都建立在三个核心概念之上：State（状态）、Node（节点）、Edge（边）。理解这三者，你就掌握了 LangGraph 的 80%。
State 是 LangGraph 的灵魂。它是一份全局共享的数据结构，图中的每一个节点都可以读取它、修改它。你可以把 State 想象成一块挂在墙上的公共黑板：
在 LangGraph 中，State 通常用 Python 的 TypedDict 来定义：
from typing import TypedDict, List, Annotatedfrom langgraph.graph.message import add_messagesimport operator# 定义一个 Stateclass AgentState(TypedDict):# 消息列表：使用 add_messages Reducer，实现追加而非覆盖messages: Annotated[List, add_messages]# 步骤计数器：使用 operator.add，每次累加step: Annotated[int, operator.add]# 用户意图：普通字段，会被直接覆盖intent: str# 是否已完成：布尔标志is_complete: bool
from typing import Annotated# 语法：Annotated[实际类型, 额外信息1, 额外信息2, ...]x: Annotated[int, "这是一个年龄字段"]
Reducer（归约器）是一个函数，它定义了“当多个节点都想修改同一个 State 字段时，如何把这些修改合并成一个最终结果”。
或者说：Reducer 决定了 State 中某个字段的更新策略——是覆盖？是追加？是累加？还是其他自定义规则？
这是 LangGraph 新手最容易踩坑的地方。默认情况下，节点返回的状态会覆盖旧状态。但在很多场景中，我们需要的是追加或累加，而不是覆盖。
比如对话消息列表：如果每个节点都返回 {"messages": [new_msg]}，默认行为会让旧消息消失。这时候就需要 Reducer 来救命。
# 不使用 Reducer：覆盖行为state["messages"] = ["你好"]# 第一次设置state["messages"] = ["世界"]# 第二次设置 → 覆盖了"你好"# 使用 add_messages Reducer：追加行为state["messages"] = ["你好"]# 第一次设置state["messages"] = ["世界"]# 第二次设置 → 结果是 ["你好", "世界"]
LangGraph 内置了几种常用的 Reducer：
add_messages：消息列表追加（最常用）operator.add：数值累加或列表拼接operator.set：集合合并你也可以自定义 Reducer 函数，实现任何你想要的合并逻辑。
Node 就是图中的一个个执行单元。每个节点都是一个普通的 Python 函数，它接收当前的 State，执行一些业务逻辑，然后返回一个状态更新字典。
# 一个最简单的节点函数def my_node(state: AgentState) -> dict:"""参数 state：当前的全局状态（只读+可写）返回值 dict：要更新的状态字段"""# 读取当前状态current_step = state.get("step", 0)last_message = state["messages"][-1] if state["messages"] else ""# 执行业务逻辑（比如调用 LLM）# response = llm.invoke(last_message)# 返回要更新的状态return {"messages": ["这是节点处理后的回复"],"step": 1,# 由于 Reducer 是 operator.add，实际效果是 step += 1}
from langgraph.graph import StateGraph# 创建图workflow = StateGraph(AgentState)# 添加节点workflow.add_node("llm_call", my_node)# 节点名称 + 节点函数workflow.add_node("tool_execute", tool_node) # 可以有多个节点workflow.add_node("format_output", format_node)
如果说 Node 是图中的“名词”，那么 Edge 就是图中的“动词”——它决定了节点之间的连接关系和执行顺序。
普通边定义的是确定性的先后顺序：A 执行完之后，一定执行 B。
# 从 START 到 NodeA（入口）workflow.add_edge(START, "llm_call")# 从 NodeA 到 NodeB（顺序执行）workflow.add_edge("llm_call", "tool_execute")# 从 NodeB 到 END（出口）workflow.add_edge("tool_execute", END)
这段代码的含义是：START → llm_call → tool_execute → END，一个标准的线性流程。
条件边是 LangGraph 真正的杀手锏。它允许你根据当前 State 的内容，动态决定下一步去哪一个节点。
# 定义路由函数：根据状态决定下一步def router_function(state: AgentState) -> str:"""返回值是目标节点的名称"""last_message = state["messages"][-1].content if state["messages"] else ""# 判断逻辑if "需要查资料" in last_message:return "search_tool" # 去搜索节点elif "需要计算" in last_message:return "calculator_tool" # 去计算节点elif "已完成" in last_message:return END # 结束else:return "llm_call"# 继续对话（循环回 LLM 节点）# 添加条件边workflow.add_conditional_edges("llm_call",# 源节点router_function, # 路由函数{# 路径映射（可选，用于校验）"search_tool": "search_tool","calculator_tool": "calculator_tool",END: END,"llm_call": "llm_call",})
条件边的威力：它让图具备了动态决策能力。同样的输入，可能走完全不同的路径；同一个节点，可以被反复访问（形成循环）。
LangGraph 预定义了两个特殊节点：
START：图的入口点。所有从外部 invoke() 传入的数据，都会首先到达 START 节点。END：图的终止点。当执行流到达 END 时，图执行完毕，返回最终 State。注意：你必须至少有一条路径从 START 出发，也必须至少有一条路径到达 END，否则编译会报错。
下面展示了从零构建一张可运行图的完整流程。
from typing import TypedDict, List, Annotatedfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessageimport operator# ===== Step 1: 定义 State =====class AgentState(TypedDict):messages: Annotated[List, add_messages]# 消息历史（追加）reflection_count: Annotated[int, operator.add]# 反思次数（累加）max_reflections: int# 最大反思次数（覆盖）# ===== Step 2: 定义 Node 函数 =====llm = ChatOpenAI(model="gpt-4o-mini")def call_model(state: AgentState) -> dict:"""调用 LLM 生成回复"""response = llm.invoke(state["messages"])return {"messages": [response]}def should_continue(state: AgentState) -> str:"""条件路由：判断是否需要继续反思"""# 如果反思次数已达上限，结束if state["reflection_count"] >= state.get("max_reflections", 3):return END# 如果最后一条消息来自 AI，检查是否需要反思last_message = state["messages"][-1]if isinstance(last_message, AIMessage):# 假设如果 AI 回复中包含"让我再想想"，则需要反思if "让我再想想" in last_message.content:return "reflect"return ENDdef reflect_node(state: AgentState) -> dict:"""反思节点：对之前的回答进行自我审查"""# 构造反思提示reflection_prompt = f"""请对你之前的回答进行反思，找出可能的不足或遗漏。如果需要改进，请给出更好的版本。原始回答：{state['messages'][-1].content}"""response = llm.invoke([HumanMessage(content=reflection_prompt)])return {"messages": [response], "reflection_count": 1}# ===== Step 3: 构建图 =====workflow = StateGraph(AgentState)# 添加节点workflow.add_node("call_model", call_model)workflow.add_node("reflect", reflect_node)# 添加边workflow.add_edge(START, "call_model") # 入口 → 调用模型workflow.add_conditional_edges( # 模型调用后，根据条件路由"call_model",should_continue,{"reflect": "reflect", END: END})workflow.add_edge("reflect", "call_model")# 反思后回到模型调用（形成循环）# ===== Step 4: 编译图 =====app = workflow.compile()# ===== Step 5: 运行图 =====initial_state = {"messages": [HumanMessage(content="帮我写一首关于秋天的诗，要有意境")],"reflection_count": 0,"max_reflections": 2,}result = app.invoke(initial_state)print(result["messages"][-1].content)